package com.marveal.rabbit.producer.task;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
import com.marveal.rabbit.producer.broker.RabbitBroker;
import com.marveal.rabbit.producer.constant.BrokerMessageStatus;
import com.marveal.rabbit.producer.entity.BrokerMessage;
import com.marveal.rabbit.producer.service.MessageStoreService;
import com.marveal.rabbit.task.annotation.ElasticJobConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;

import java.util.List;

@ElasticJobConfig(
        jobName = "com.marveal.rabbit.producer.task.RetryMessageDataflowJob",
        cron = "0/10 * * * * ?",
        description = "可靠性投递消息补偿任务",
        overwrite = true,
        shardingTotalCount = 1
)
public class RetryMessageDataflowJob implements DataflowJob<BrokerMessage> {

    private static final Logger LOGGER = LoggerFactory.getLogger(RedisAutoConfiguration.class);

    private final MessageStoreService messageStoreService;

    private final RabbitBroker rabbitBroker;

    public RetryMessageDataflowJob(MessageStoreService messageStoreService, RabbitBroker rabbitBroker) {
        this.messageStoreService = messageStoreService;
        this.rabbitBroker = rabbitBroker;
    }

    private static final int MAX_RETRY_COUNT = 3;

    @Override
    public List<BrokerMessage> fetchData(ShardingContext shardingContext) {
        List<BrokerMessage> brokerMessages = messageStoreService.fetchTimeOutMessage4Retry(BrokerMessageStatus.SENDING);
        LOGGER.info("#RetryMessageDataflowJob.fetchData# fetch message: {}", brokerMessages.size());
        return brokerMessages;
    }

    @Override
    public void processData(ShardingContext shardingContext, List<BrokerMessage> dates) {
        dates.forEach(message -> {
            String messageId = message.getMessageId();
            if (message.getTryCount() >= MAX_RETRY_COUNT) {
                messageStoreService.failure(messageId);
                LOGGER.warn("#RetryMessageDataflowJob.processData# message {} retry {} times, failure!",
                        messageId, message.getTryCount());
                return;
            }
            // 更新tryCount
            messageStoreService.updateTryCount(messageId);

            // 重发消息
            rabbitBroker.reliantSend(message.getMessage());
        });


    }
}
